package com.test.study.ToolsUtils.strom.workdsum;

import java.io.File;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.Collection;
import java.util.List;
import java.util.Map;

import org.apache.commons.io.FileUtils;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

@SuppressWarnings("serial")
public class GoloabSpout extends BaseRichSpout {
	@SuppressWarnings({ "unused", "rawtypes" })
	private Map conf;
	@SuppressWarnings("unused")
	private TopologyContext context;
	private SpoutOutputCollector collector;
	private static String filePath="/Users/huyong/App/file/testDir";
	

	@SuppressWarnings("rawtypes")
	@Override
	public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {

		this.conf=conf;
		this.collector=collector;
		this.context=context;
	}

	@Override
	public void nextTuple() {
		//监控指定目录下文件的变化
		Collection<File> listFiles = FileUtils.listFiles(new File(filePath), new String[] {"txt"}, true);
		if(listFiles!=null && listFiles.size()>0) {
			listFiles.forEach(c->{
				try {
					List<String> readLines = FileUtils.readLines(c.getAbsoluteFile(),Charset.defaultCharset());
					if(readLines!=null && readLines.size()>0) {
						readLines.forEach(line->{
							this.collector.emit(new Values(line));
						});
					}
				} catch (IOException e) {
					e.printStackTrace();
				}
				//rm file
				try {
					FileUtils.moveFile(c, new File(c.getAbsolutePath() + System.currentTimeMillis()));
				} catch (IOException e) {
					e.printStackTrace();
				}
			});
		}

	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("line"));
	}

}
